package cn.zx.rmq.producer;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.ArrayList;
import java.util.List;

/**
 * @Package: cn.zx.rmq.consumer
 * @ClassName: PartialOrderProducer
 * @Author: Think
 * @CreateTime: 2021/6/28 15:08
 * @Description:所有消息单线程顺序发送，并且在消息队列中按照发送顺序存储在同一个队列中
 */
@Slf4j
public class GlobalOrderProducer {

    @SneakyThrows
    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer("GlobalOrderProducer-Group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        // 构建消息集合，其中第三个参数keys是业务标识，表示同一个人的ID
        List<Message> messageList = new ArrayList<>();
        Message msg1 = new Message("KYC", "about-order", "321", ("zhangsan下单了").getBytes(RemotingHelper.DEFAULT_CHARSET));
        Message msg2 = new Message("KYC", "about-order", "521", ("lisi下单了").getBytes(RemotingHelper.DEFAULT_CHARSET));
        Message msg3 = new Message("KYC", "about-order", "321", ("zhangsan收货了").getBytes(RemotingHelper.DEFAULT_CHARSET));
        Message msg4 = new Message("KYC", "about-order", "521", ("lisi收货了").getBytes(RemotingHelper.DEFAULT_CHARSET));
        Message msg5 = new Message("KYC", "about-order", "521", ("lisi评价了").getBytes(RemotingHelper.DEFAULT_CHARSET));
        Message msg6 = new Message("KYC", "about-order", "321", ("zhangsan评价了").getBytes(RemotingHelper.DEFAULT_CHARSET));
        messageList.add(msg1);
        messageList.add(msg2);
        messageList.add(msg3);
        messageList.add(msg4);
        messageList.add(msg5);
        messageList.add(msg6);

        for (Message msg : messageList) {
            // 将使用SelectMessageQueueByHash队列选择器，根据客户ID（keys）来决定选择同一个客户使用哪一个队列
            SendResult sendResult = producer.send(msg, new SelectMessageQueueByHash(), 0);
            log.info("当前消息发送结果为：{}", sendResult);
        }

        producer.shutdown();
        log.info("所有消息发送完毕！");
    }

}
